#include <sys/eventfd.h>
#include <unistd.h> // read()
// #include <fcntl.h>
// #include <errno.h>
// #include <memory>

#include "./EventLoop.h"
#include "../thread/CurrentThread.h"
#include "./Poller.h"
#include "../utility/Logger.h"
#include "./Channel.h"

// 防止一个线程创建多个EventLoop
// thread中一个loop创建就指向loop
// 再创建loop该指针不为空就失败了
// OPTIMIZE: __thread -> thread_local
// __thread EventLoop *t_loopInThisThread = nullptr;
thread_local EventLoop *kThreadLoopAddr = nullptr;

// Poller IO复用接口默认超时时间
const int kPollTimeMs = 10000; // 10000ms = 10s

/* 创建线程之后主线程和子线程谁先运行是不确定的。
 * 通过一个eventfd在线程之间传递数据的好处是多个线程无需上锁就可以实现同步。
 * eventfd支持的最低内核版本为Linux 2.6.27,在2.6.26及之前的版本也可以使用eventfd，但是flags必须设置为0。
 * 函数原型：
 *     #include <sys/eventfd.h>
 *     int eventfd(unsigned int initval, int flags);
 * 参数说明：
 *      initval,初始化计数器的值。
 *      flags, EFD_NONBLOCK,设置socket为非阻塞。
 *             EFD_CLOEXEC，执行fork的时候，在父进程中的描述符会自动关闭，子进程中的描述符保留。
 *             × 父进程有，子进程无
 * 场景：
 *     eventfd可以用于同一个进程之中的线程之间的通信。
 *     eventfd还可以用于同亲缘关系的进程之间的通信。
 *     eventfd用于不同亲缘关系的进程之间通信的话需要把eventfd放在几个进程共享的共享内存中（没有测试过）。
 */
// 没有用生产者消费者模型队列缓冲隔开，相对没那么好处理，而是用wakeup fd唤醒通信
// 创建wakeupfd 用来notify唤醒subReactor处理新来的channel
int createEventfd()
{
    int eventfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (eventfd < 0)
    {
        LOG_FATAL("eventfd error:%d\n", errno); // 失败就无法通知了是致命错误
    }

    return eventfd;
}

EventLoop::EventLoop()
    : tid_(CurrentThread::tid()),
      isStart_(false),
      isQuit_(false),
      poller_(Poller::getDefaultPoller(this)),
      wakeupFd_(createEventfd()),
      wakeupChan_(new Channel(this, wakeupFd_)),
      hasPendingCallbacks_(false)
{
    LOG_DEBUG("EventLoop created %p in thread %d\n", this, this->tid_);

    if (kThreadLoopAddr)
    {
        LOG_FATAL("Another EventLoop %p exists in this thread %d\n", kThreadLoopAddr, this->tid_);
    }
    else
    {
        kThreadLoopAddr = this;
    }

    // 这里是构造函数，每个loop构造都要设置一个这个为了让其他loop唤醒
    // 最开始就是读事件，读事件不重要
    // 重点就是mainreactor write()唤醒相应的subreactor read()
    // 然后处理业务，子loop拿到主loop给的channel

    // 注意顺序：先注册回调，再设置状态
    // 设置wakeupfd的事件类型以及发生事件后的回调操作
    this->wakeupChan_->setReadCallback(std::bind(&EventLoop::handleWakeupFdRead, this));
    // 每一个eventloop都将监听wakeupchannel的EPOLLIN读事件
    this->wakeupChan_->enableReadEvent();
}

void EventLoop::handleWakeupFdRead()
{
    // 约定读取8字节，不重要
    uint64_t one{1};
    ssize_t n = read(this->wakeupFd_, &one, sizeof(one));
    if (n != sizeof(one))
    {
        LOG_ERROR("EventLoop::handleWakeupFdRead() reads %lu bytes instead of 8\n", n);
    }
}

EventLoop::~EventLoop()
{
    this->wakeupChan_->disableAllEvent();
    this->wakeupChan_->removeChannel();

    ::close(this->wakeupFd_);

    kThreadLoopAddr = nullptr;
}

void EventLoop::start()
{
    LOG_INFO("EventLoop %p start looping\n", this);

    this->isStart_ = true;
    this->isQuit_ = false;

    while (!this->isQuit_)
    {
        this->activeChans_.clear();
        // 监听三类fd：wakeupFd，listenFd，connFd
        this->pollReturnTime_ = this->poller_->poll(kPollTimeMs, &this->activeChans_);

        // 1. 唤醒处理就绪事件
        // Poller就绪事件activeChans_给loop，loop再让channel处理
        for (Channel *chan : this->activeChans_)
        {
            chan->handleEvent(this->pollReturnTime_);
        }

        // 2. 执行注册给该loop的回调
        // mainloop有新连接TcpConnection
        // 给subloop注册回调TcpConnection::connectEstablished
        // 唤醒subloop处理TcpConnection
        this->doPendingCallbacks();
    }

    this->isStart_ = false;

    LOG_INFO("EventLoop %p stop looping.\n", this);
}

void EventLoop::doPendingCallbacks()
{
    this->hasPendingCallbacks_ = true;
    std::vector<Callback> cbs{};

    // 如果不把回调swap()转移，需要上锁不断执行回调，其他loop注册新的回调会因为锁阻塞
    // 回调可以转移后慢慢执行
    // 减小了锁范围
    // 提升效率
    // 避免死锁：在临界区cb()如果又调用queueInLoop()（该函数里又尝试加锁this->mutex_而失败阻塞）
    {
        std::unique_lock<std::mutex> lock(this->mutex_);
        cbs.swap(this->pendingCallbacks_);
    }

    for (const Callback &cb : cbs)
    {
        cb(); // 执行当前loop需要执行的回调操作
    }

    this->hasPendingCallbacks_ = false;
}

void EventLoop::quit()
{
    // 1. 线程a执行loop a的事件循环和poll等核心操作（线程a运行loop a）
    // 2. 线程a也可以获取到loop b向其提交任务（线程a运行loop b）

    // 1. 线程a调用loop a的quit()，肯定不poll阻塞，正常退出
    // 2. 线程a调用loop b的quit()，loop b核心操作实际需要线程b处理，唤醒线程b
    this->isQuit_ = true;

    if (!this->isLoopInThread())
    {
        this->wakeupLoopThread();
    }
}

bool EventLoop::isLoopInThread() const
{
    return this->tid_ == CurrentThread::tid();
}

void EventLoop::wakeupLoopThread()
{
    // 写什么不重要，wakeupChannel读事件解阻塞唤醒当前loop线程
    uint64_t one = 1;
    ssize_t n = ::write(this->wakeupFd_, &one, sizeof(one));
    if (n != sizeof(one))
    {
        LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8\n", n);
    }
}

Timestamp EventLoop::getPollReturnTime() const
{
    return this->pollReturnTime_;
}

void EventLoop::runCallback(Callback &&cb)
{
    // 1. 线程a loop a，直接执行回调
    // 2. 线程b loop a，唤醒线程a执行回调
    if (this->isLoopInThread())
    {
        cb();
    }
    else
    {
        this->queueInLoop(std::move(cb)); // 需要移动
    }
}

void EventLoop::queueInLoop(Callback &&cb)
{
    {
        std::unique_lock<std::mutex> lock(this->mutex_);
        this->pendingCallbacks_.emplace_back(cb);
    }

    // 如果
    // 1. 当前loop不在所在线程
    // 2. hasPendingCallbacks_：当前loop正在执行回调，但是loop又有了新的回调
    // 唤醒loop所在线程/写数据
    // loop下一次poll()不再阻塞（阻塞的话会延迟前一次新加入的回调的执行）
    // 继续执行新加入的回调
    if (!this->isLoopInThread() || this->hasPendingCallbacks_)
    {
        this->wakeupLoopThread();
    }
}

bool EventLoop::hasChannel(Channel *chan)
{
    return this->poller_->hasChannel(chan);
}

void EventLoop::updateChannel(Channel *chan)
{
    this->poller_->updateChannel(chan);
}

void EventLoop::removeChannel(Channel *chan)
{
    this->poller_->removeChannel(chan);
}